译文 | Pulsar Functions 简要指南：原理、目标与规划

原文链接：https://www.splunk.com/en_us/blog/it/introducing-pulsar-functions.html

本文作者为 Jerry Peng，Splunk 首席软件工程师。

---

阅读本文大约需要 6 分钟。

Pulsar Functions 是 Apache Pulsar 的一个轻量级计算框架，也是一个类似于 AWS Lambda、Google Cloud Functions 等的 FAAS 平台。Pulsar Functions 的设计灵感来自于 Heron 这样的流处理引擎，Pulsar Functions 将会拓展 Pulsar 和整个消息领域的未来。使用 Pulsar Functions，用户可以轻松地部署和管理 function，通过 function 从 Pulsar topic 读取数据或者生产新数据到 Pulsar topic。

近来，Pulsar 在消息领域受到广泛关注，因为 Pulsar 是唯一结合了队列和流的开源平台，原生支持多租户，支持开箱即用的多数据中心跨机房复制，还支持 effectively-once 消息传递语义保证等。

引入 Pulsar Functions 后，Pulsar 成为统一的消息投递/计算/存储平台。只需部署一套 Pulsar 集群，便可以实现一个计算引擎，页面简单，操作便捷。

 Pulsar Functions 工作原理 

使用 Pulsar Functions，您可以用 Java、Python、Go 语言编写 functions，部署简单。编写模式有以下两种：

• 在本地运行模式下，用户可以在集群外部，即本地运行 Pulsar Functions。

• 在集群模式下，用户可以在集群内部运行 Pulsar Functions。

在集群模式下部署 Functions 时，Apache BookKeeper 会自动处理状态存储。

 为什么使用 Pulsar Functions？

目前，Pulsar 可以与任一计算或流处理引擎（Stream Processing Engine，SPE）一起使用，如 Heron。而 Pulsar Functions 的优点在于无需部署就可以直接使用 SPE。

无需配置整个流处理集群，Pulsar Functions 即可简单处理消息总线中的数据。Pulsar Functions 使用的 极其简单，用户可以消费 Pulsar topic 上的消息，使用定制逻辑处理数据，并在必要时将结果写回到其他 Pulsar topic 中。例如，使用 Java 的用户可以轻松实现 java.util.function.Function 接口，此接口只有一个 apply 方法。示例如下：

import java.util.Function;

public class ExclamationFunction implements Function<String, String> {
    @Override
    public String apply(String input) { return String.format("%s!", input); }
}

实现 Pulsar Functions 非常容易，甚至不需要 SDK。如果用户需要 context 信息，例如 function 名称，用户只需实现 PulsarFunctions 接口即可，而无需实现 Java Function 接口。示例如下：

public interface PulsarFunction<I, O> {
    O process(I input, Context context) throws Exception;
}

这就实现了 Pulsar Functions。

从 API 可以看出，我们决定将这个框架命名为 Pulsar Functions 是因为它就像一个函数一样，非常简单明了。在“本地运行”模式下使用 Pulsar Functions 时，用户可以自行决定在哪里运行 function（例如，在笔记本电脑上）。如果用户向 Pulsar 集群提交了 Pulsar Functions，集群将负责运行这些 function。后文中会详细介绍 Pulsar Functions 在两种不同模式下的运行。

接下来，我们来深入了解一下 Pulsar Functions 的基本信息、研发背景以及运行时的整体概况。

 为什么开发 Pulsar Functions？

在消息领域从业多年，经验告诉我们绝大部分数据处理应用场景都是简单、轻量级的，比如：

• 简单的 ETL（提取、转化、加载）

• 操作基于事件的服务

• 实时聚合

• 事件路由

Pulsar Functions 不仅能处理上述用例，还大大简化了部署，因而在减少了开发时间的同时，实现了开发人员产出的最大化。根据业内用户的反馈，我们认为操作简单很重要，并总结了一些痛点问题：

1. 用户希望处理消息总线中的数据操作简单。单独部署一套完整的 SPE 集群操作复杂，任务量繁重。事实上，用户只需要 SPE 功能的一小部分。

2. 完整的 SPE 有许多功能，因此部署、监控、运维等操作会相对复杂。如果用户想简单处理数据，SPE 的某些操作代价并不合理。

3. 一些用户认为，对于简单用例而言，SPE 比较完善，其 API 过于复杂，且操作不便。很多 SPE 都具有基于功能编程模型的 API，如 map、flatmap、reduce 等。这些 SPE 的 API 功能强大，但对于多数用例来说，尤其是在用户不太习惯功能性编程范式时，这些 API 过于复杂。

 设计目标 

考虑到上述问题，我们制定了 Pulsar Functions 的设计目标：

鉴于上述设计目标，我们开发了 Pulsar Functions 运行时。

 运行时概况 

由执行程序，即实例运行 Pulsar Functions。单个实例执行 function 的一个副本。一个 function 可以有多个实例，实现了 Pulsar Functions 的内置并行性，可以在 function 的配置中设置实例数量。

为了最大化部署灵活性，Pulsar Functions 合并了多个执行环境和运行时，原生支持以下运行环境（以后会支持更多运行环境）：

每个执行环境的成本各不相同，所提供的隔离保证也不同。

除了表格中的运行时，Pulsar Functions 还可以提供许多运行时，用于执行以不同编程语言编写的 function。Pulsar Functions 目前仅支持 Java 、Python 和 Go，以后会支持更多语言。

图 1. Pulsar Functions 运行时图解

Pulsar Functions 可以从一个或多个 Pulsar topic 消费消息，并且可以选择是否将结果写回到另一个 Pulsar topic。可以由一个或多个 Pulsar function 执行用户自定义的处理逻辑。Pulsar Functions 可以通过 context 接口保留中间结果，其他用户可以查询 function 状态来提取这些结果。

 运行时概况 

运行 Pulsar Functions 最简单的方式是实例化运行时和 function，并在本地运行二者（本地运行模式）。命令行工具可以简化这一操作。在本地运行模式下，function 作为独立的运行时运行，可通过任何可用的进程、Docker 容器、线程控制机制等对其进行监控。用户可以在计算机上手动生成这些运行时，也可以使用高级调度程序，如 Mesos/Kubernetes 在集群中进行分配。本地运行模式下，启动单个 Pulsar Function 的命令示例如下：

$ bin/pulsar-admin functions localrun \
  --inputs persistent://sample/standalone/ns1/test_src \
  --output persistent://sample/standalone/ns1/test_result \
  --jar examples/api-examples.jar \
  --className org.apache.pulsar.functions.api.examples.ExclamationFunction

用户还可以在 Pulsar 集群内部与 broker 一起运行 function。在这种模式下，用户可以将多个 function “提交”到正在运行的 Pulsar 集群，Pulsar 将负责跨集群分发、监听、执行这些 function。这种模式允许用户专注于开发 function 而无需担心 function 生命周期的管理。向 Pulsar 集群提交单个待运行 Pulsar Function 的示例如下：

$ bin/pulsar-admin functions create \
  --inputs persistent://sample/standalone/ns1/test_src \
  --output persistent://sample/standalone/ns1/test_result \
  --jar examples/api-examples.jar \
  --className org.apache.pulsar.functions.api.examples.ExclamationFunction \
  --name myFunction

还可以将 function 的全部配置放在 YAML 文件中，如下所示：

inputs: persistent://sample/standalone/ns1/test_src
output: persistent://sample/standalone/ns1/test_result
jar: examples/api-examples.jar
className: org.apache.pulsar.functions.api.examples.ExclamationFunction
name: myFunction

如果想通过 YAML 文件配置 function，使用 create 命令会方便得多：

$ bin/pulsar-admin functions create \
  --configFile ./my-function-config.yaml

 处理保证 

Pulsar Functions 提供以下处理语义，可以通过不同的 function 分别指定这三种语义：

• At most once

• At least once

• Effectively once

关于这三种处理语义的更多详细信息，参阅 Apache Pulsar 的 exactly-once 语义：

https://www.splunk.com/en_us/blog/it/effectively-once-semantics-in-apache-pulsar.html

结合使用 at-least-once 和服务器端的消息去重就可以实现 effectively-once 处理。也就是说，可以进行两次状态更新，但只会应用其中一次，重复的状态都将在服务器端丢弃。

 Pulsar Functions 开发规划 

在接下来的几个月里，Pulsar Functions 可能会原生支持以下性能：

• 扩展语言支持：目前 Pulsar Functions 只支持 Java、Python 和 Go，接下来会支持 JavaScript。由于缺少原生的 Pulsar 客户端，Pulsar Functions 支持的语言也受到限制。包装现有功能齐全的 C++ 客户端是扩展语言支持的一种简单方法。

• Function 组合：Funtion 组合将会简化指定整个 function 管道的操作，实现方法很多。例如：为 DSL 添加 map、flatMap、reduce、join 等操作符；或使用 JSON 或 YAML 将多个 function 组合在一起等。

• Schema Registry 集成：Pulsar 2.0 新增特性中最令人期待的无疑是 Schema Registry。可以通过提交事件检查，或通过需要序列化和反序列化（SerDe）的类，大致推断出序列化和反序列化后类的形式与 Schema Registry 集成。

• 与成熟的调度程序集成：用户可以轻松提交、管理要在 Mesos、Kubernetes、Nomad、Yarn 等上运行的 function。

 结语 

本文详细介绍了 Pulsar Functions 的开发背景、开发目标、运行时等信息。感谢 Jerry Peng、Sanjeev Kulkarmi，以及 Sijie Guo 对本文撰写提供的帮助与支持。

---

想和 Pulsar 社区大咖谈笑风生吗？那就快来报名参加 Pulsar Summit Asia 2020 吧！限量免费门票，先到先得，扫描下方二维码、或者点击阅读原文都可以通过活动行平台报名哦☟

👍 相关阅读

点击预约，限量名额，第一时间获取大会视频及PDF资料 ☟